/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.sdk.io.redis;



import com.google.auto.value.AutoValue;

import com.bff.gaia.unified.sdk.transforms.display.DisplayData;

import redis.clients.jedis.Jedis;

import redis.clients.jedis.Protocol;



import javax.annotation.Nullable;

import java.io.Serializable;



import static com.bff.gaia.unified.vendor.guava.com.google.common.base.Preconditions.checkArgument;



/**

 * {@code RedisConnectionConfiguration} describes and wraps a connectionConfiguration to Redis

 * server or cluster.

 */

@AutoValue

public abstract class RedisConnectionConfiguration implements Serializable {



  abstract String host();



  abstract int port();



  @Nullable

  abstract String auth();



  abstract int timeout();



  abstract boolean ssl();



  abstract Builder builder();



  @AutoValue.Builder

  abstract static class Builder {

    abstract Builder setHost(String host);



    abstract Builder setPort(int port);



    abstract Builder setAuth(String auth);



    abstract Builder setTimeout(int timeout);



    abstract Builder setSsl(boolean ssl);



    abstract RedisConnectionConfiguration build();

  }



  public static RedisConnectionConfiguration create() {

    return new AutoValue_RedisConnectionConfiguration.Builder()

        .setHost(Protocol.DEFAULT_HOST)

        .setPort(Protocol.DEFAULT_PORT)

        .setTimeout(Protocol.DEFAULT_TIMEOUT)

        .setSsl(false)

        .build();

  }



  public static RedisConnectionConfiguration create(String host, int port) {

    return new AutoValue_RedisConnectionConfiguration.Builder()

        .setHost(host)

        .setPort(port)

        .setTimeout(Protocol.DEFAULT_TIMEOUT)

        .setSsl(false)

        .build();

  }



  /** Define the host name of the Redis server. */

  public RedisConnectionConfiguration withHost(String host) {

    checkArgument(host != null, "host can not be null");

    return builder().setHost(host).build();

  }



  /** Define the port number of the Redis server. */

  public RedisConnectionConfiguration withPort(int port) {

    checkArgument(port > 0, "port can not be negative or 0");

    return builder().setPort(port).build();

  }



  /** Define the password to authenticate on the Redis server. */

  public RedisConnectionConfiguration withAuth(String auth) {

    checkArgument(auth != null, "auth can not be null");

    return builder().setAuth(auth).build();

  }



  /**

   * Define the Redis connection timeout. A timeout of zero is interpreted as an infinite timeout.

   */

  public RedisConnectionConfiguration withTimeout(int timeout) {

    checkArgument(timeout >= 0, "timeout can not be negative");

    return builder().setTimeout(timeout).build();

  }



  /** Enable SSL connection to Redis server. */

  public RedisConnectionConfiguration enableSSL() {

    return builder().setSsl(true).build();

  }



  /** Connect to the Redis instance. */

  public Jedis connect() {

    Jedis jedis = new Jedis(host(), port(), timeout(), ssl());

    if (auth() != null) {

      jedis.auth(auth());

    }

    return jedis;

  }



  /** Populate the display data with connectionConfiguration details. */

  public void populateDisplayData(DisplayData.Builder builder) {

    builder.add(DisplayData.item("host", host()));

    builder.add(DisplayData.item("port", port()));

    builder.addIfNotNull(DisplayData.item("timeout", timeout()));

    builder.add(DisplayData.item("ssl", ssl()));

  }

}